﻿#include "ThreadPool.h"
#include <functional>
using namespace std;
template <typename T>
ThreadPool<T>::ThreadPool(int min, int max)
{
	//实例化
	taskQ = new TaskQueue<T>;
	this->minNum = min;
	this->maxNum = max;
	this->busyNum = 0;
	this->liveNum = 0;
	this->exitNum = 0;
	this->shutdown = false;
	//创建线程
	ThreadPool* pool = this;
	//直接通过传递参数构造出对象，比push_back效率要高  
	mangerThread.emplace_back(ThreadPool::manager,this);
	mangerThread[0].detach();
	 
	for (int i = 0; i < min;)
	{
		uint64_t localTime = pool->getCurrentTime();
		//防止出现重复时间
		if (workerThreads.find(localTime) != workerThreads.end())
		{
			continue;
		}
		pool->workerThreads.emplace(localTime, thread(ThreadPool<T>::worker, pool, localTime));
		pool->workerThreads[localTime].detach();
		i++;
    }
	

}
template <typename T>
ThreadPool<T>::~ThreadPool()
{
	mutexPool.lock();
	//结束线程
	shutdown = true;
	mutexPool.unlock();
	
	//唤醒阻塞线程退出
	while (true)
	{
		mutexPool.lock();
		int num = liveNum;
		if(num > 0)
			notEmpty.notify_one();
		mutexPool.unlock();

		cout << "current thread count:" << num<<endl;
		if (num <= 0)
			break;
		std::this_thread::sleep_for(std::chrono::milliseconds(500));
	}

	mangerThread.clear();
	workerThreads.clear();
	if (taskQ)
	{
		delete taskQ;
		taskQ = nullptr;
	}
}
template <typename T>
void ThreadPool<T>::addTask(Task<T> task)
{
	if (shutdown || !taskQ)
		return;
	//往任务队列添加任务并且唤醒工作线程
	taskQ->addTask(task);
	notEmpty.notify_one();

}
template <typename T>
int ThreadPool<T>::getBusyNum()
{
	std::unique_lock<std::mutex> locker(mutexPool);
	return busyNum;
}
template <typename T>
int ThreadPool<T>::getAliveNum()
{
	std::unique_lock<std::mutex> locker(mutexPool);
	return liveNum;
}
template <typename T>
void* ThreadPool<T>::worker(void* arg, uint64_t localtime)
{
	ThreadPool<T>* pool = static_cast<ThreadPool<T>*>(arg);
	if (!pool)
		return nullptr;
	
	pool->mutexPool.lock();
	pool->liveNum++;
	pool->mutexPool.unlock();
	
	
	while (true)
	{
		{
			//判断当前任务队列是否为空
			std::unique_lock<std::mutex> locker(pool->mutexPool);
			while (pool->taskQ && pool->taskQ->taskNumber() == 0 && !pool->shutdown)
			{
				//阻塞工作线程
				pool->notEmpty.wait(locker, [=]() {return (pool->taskQ->taskNumber() > 0 || pool->exitNum > 0 || pool->shutdown); });
				//判断是否要销毁线程
				if (pool->exitNum > 0)
				{
					pool->exitNum--;
					if (pool->liveNum > pool->minNum)
					{
						pool->liveNum--;
						pool->threadExit(localtime);
						return nullptr;
					}
				}

				if (pool->shutdown)
				{
					pool->liveNum--;
					pool->threadExit(localtime);
					return nullptr;
				}
			}
		}
		

		pool->mutexPool.lock();
		//判断线程池是否被关闭了
		if (pool->shutdown)
		{
			pool->liveNum--;
			pool->threadExit(localtime);
			pool->mutexPool.unlock();
			return nullptr;
		}
		pool->busyNum++;
		pool->mutexPool.unlock();

		if (pool->taskQ->taskNumber() > 0)
		{
			Task<T> task = pool->taskQ->takeTask();
			task.f(task.arg);
			delete task.arg;
		}

		pool->mutexPool.lock();
		pool->busyNum--;
		pool->mutexPool.unlock();
	}

	return nullptr;
}

template <typename T>
void* ThreadPool<T>::manager(void* arg)
{
	ThreadPool<T>* pool = static_cast<ThreadPool<T>*>(arg);
	if (!pool)
		return nullptr;
	while (true)
	{
		//每隔3s检测一次
		std::this_thread::sleep_for(std::chrono::milliseconds(3000));
		//取出线程池中任务数量和当前线程数量

		std::unique_lock<std::mutex> locker(pool->mutexPool);
		if (pool->shutdown)
			return nullptr;
		int queueSize = pool->taskQ->taskNumber();
		int liveNum = pool->liveNum;
		int busyNum = pool->busyNum;

		//批量添加线程(任务个数 > 存活的线程数 && 存活的线程数 < 最大)
		if (queueSize > liveNum && liveNum < pool->maxNum)
		{
			for (int i = 0; i < 2; )
			{
				uint64_t localTime = pool->getCurrentTime();
				if (pool->workerThreads.find(localTime) != pool->workerThreads.end())
				{
					continue;
				}
				pool->workerThreads.emplace(localTime, thread(ThreadPool<T>::worker, pool, localTime));
				pool->workerThreads[localTime].detach();
				i++;
				cout << "current thread size:" << pool->workerThreads.size()<<endl;
			}
		}
		//批量结束线程  忙线程*2 < 存活的线程 && 存活的线程 > 最小线程数
		else if (busyNum * 2 < liveNum && liveNum > pool->minNum)
		{
			pool->exitNum = 2;
			for (int i = 0; i < 2; i++)
				pool->notEmpty.notify_one();
		}
	}	
	return nullptr;
}
template <typename T>
void ThreadPool<T>::threadExit(uint64_t localtime)
{
	for(auto it = workerThreads.begin();it!=workerThreads.end();it++)
	{
		if (it->first == localtime)
		{
			cout << std::this_thread::get_id() << "  " << "thread exit!!!" << endl;
			workerThreads.erase(it);
			break;
		}
	}
}

template<typename T>
uint64_t ThreadPool<T>::getCurrentTime()
{
	struct timeb tv;
	ftime(&tv);
	return tv.time * 1000 + tv.millitm;
}
